package me.chendan

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * Created by chendan on 16-4-5.
  */
class ReadingFromNc {

}

object ReadingFromNc extends App {
  def updateSum(values: Seq[Double], state: Option[Double]) = {
    Some(state.getOrElse(0.0) + values.sum)
  }

  val conf = new SparkConf().setAppName("UserCostCalculator")
  val ssc = new StreamingContext(conf, Seconds(1))
  ssc.checkpoint("/tmp")
  val lines = ssc.socketTextStream("localhost", 9999)
  val logStream = lines.map(line => WebUtil.paseFromLines(line))
  val logWindow = logStream.window(Seconds(5), Seconds(3))
  val pagePairStream = logWindow.map(entry => (entry.uid, entry.cost))
  val pageCountStream = pagePairStream.updateStateByKey(updateSum _)

  pageCountStream.print()

  ssc.start()
  ssc.awaitTermination()
}
